package ruc.InfluxDB.write;

import okhttp3.*;
import ruc.Consts.constant;
import ruc.utils.IOUtils;
import ruc.InfluxDB.InfluxDB_Common;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

public class InfluxDB_ConcurrentWrite {
    private String writeURL = "";
    private String tbName = "";
    private String delURL = "http://%s:%s/query?pretty=true&db=%s&q=drop measurement %s";

    MediaType MEDIA_TYPE_TEXT = MediaType.parse("text/plain");

    private ReentrantLock lock = new ReentrantLock();

    public InfluxDB_ConcurrentWrite(String IP, String port, String tbName, String dbName) {
        String baseUrl = String.format("http://%s:%s", IP, port);
        this.writeURL = baseUrl + "/write?precision=ms&db=" + dbName; //Write timestamp in ms
        this.tbName = tbName;
        this.delURL = String.format(delURL, IP, port, dbName, tbName);
    }

    public long splitAndSend(String values[], int all, int num){
        lock.lock();
        try{
            StringBuffer sb = new StringBuffer();
            for(int i = 0; i < values.length; i++) {
                if (i % all != num) {
                    continue;
                }

                String[] v = values[i].split(constant.Delimiter);
                String eSQL = String.format("%s,userInfo=%s,server=%s,hostname=%s cpu_usage=%s,disk_usage=%s,memory_usage=%s %s", tbName, v[1], v[2], v[3], v[4], v[5], v[6], v[0]);
                sb.append(eSQL);
                sb.append("\n");
            }

            Request request = new Request.Builder()
                    .url(writeURL)
                    .post(RequestBody.create(MEDIA_TYPE_TEXT, sb.toString()))
                    .build();

            return InfluxDB_Common.sendRequest(request);
        } finally {
            lock.unlock();
        }
    }

    public void MulInsertData(String[] values, int threadNum[]) throws InterruptedException, IOException {

        for(int i = 0; i < threadNum.length; i++) {
            AtomicLong costTime = new AtomicLong();
            CountDownLatch countDownLatch = new CountDownLatch(i);

            for(int j = 0; j < threadNum[i]; j++) {
                int finalI = i;
                int finalJ = j;
                new Thread(()->{
                    costTime.addAndGet(splitAndSend(values, threadNum[finalI], finalJ));
                    countDownLatch.countDown();
                }).start();
            }

            countDownLatch.await();
            System.out.printf("[TSBench-D] [InfluxDB_ConcurrentWrite]>>>>>> threadNum %d, insert time %d ms >>>>>> \n", threadNum[i], costTime.longValue());
            IOUtils.WriteData(constant.GetWriteFilePath("InfluxDB", "ConcurrentWrite"),
                    new String[]{String.valueOf(System.nanoTime()), String.valueOf(threadNum[i]), String.valueOf(costTime.longValue())});

            // clear data
            Request request = new Request.Builder()
                    .url(delURL)
                    .method("POST", RequestBody.create(MEDIA_TYPE_TEXT, ""))
                    .build();
            Response response = InfluxDB_Common.getClient().newCall(request).execute();
            response.close();
        }
    }

    public static void main(String[] args) throws InterruptedException, IOException {
        InfluxDB_ConcurrentWrite connect = new InfluxDB_ConcurrentWrite(constant.IP, constant.InfluxDB_Port,constant.Concurrent_tbName, constant.dbName);
        connect.MulInsertData(IOUtils.GetData(), constant.ConWrite_Thread);
    }
}
